package cn.linkai.hadoop.invertedindex;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

public class InvertedIndexTwiceMRSecond {

    public static class InvertedIndexTwiceMRSecondMapper extends Mapper<LongWritable,Text,Text,Text>{

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer tokenizer=new StringTokenizer(value.toString());
            String[] temp=tokenizer.nextToken().split("->");
            String word=temp[0];
            String fileAndCount=temp[1]+"->"+tokenizer.nextToken();
            context.write(new Text(word),new Text(fileAndCount));
        }
    }

    public static class InvertedIndexTwiceMRSecondReducer extends Reducer<Text,Text,Text,Text>{
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            Iterator<Text> it=values.iterator();
            StringBuffer buffer=new StringBuffer();
            while(it.hasNext()){
                Text text=it.next();
                buffer.append(text.toString());
                buffer.append(",");
            }
            buffer.deleteCharAt(buffer.length()-1);
            context.write(key,new Text(buffer.toString()));
        }
    }
}
